package com.hanrt.mr.nline;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 1.一旦启动这个线程，运行Job
 * 2.本地模式主要用于测试程序是否正确！
 * 3. 报错：
 * 	ExitCodeException exitCode=1: /bin/bash: line 0: fg: no job control
 *
 * 	NLineInputFormat
 * 	切片： 读取配置中mapreduce.input.lineinputformat.linespermap，默认为1，以文件为单位，切片默认每行作为1片！
 * 	RecordReader:  LineRecordReader,一次处理一行，将一行内容的偏移量作为key，一行内容作为value!
 * 			LongWritable	key
 * 			Text value
 * 适合文件的每一行特别长，处理逻辑又复杂的情况。
 */
public class WCDriver {
	
	public static void main(String[] args) throws Exception {
		
		Path inputPath=new Path("F:\\idea2019_workspace\\bigdata202006_projects\\hadoopopr\\inputdata\\nline");
		Path outputPath=new Path("F:\\idea2019_workspace\\bigdata202006_projects\\hadoopopr\\inputdata\\nline\\result");
	
		//作为整个Job的配置
		Configuration conf = new Configuration();
		// 设置为两行一片 两个文件,第一个文件为5行 分为3片,第二个文件为3行 分为2片 一共分了5片
		conf.set("mapreduce.input.lineinputformat.linespermap", "2");
		
		
		//保证输出目录不存在
		FileSystem fs=FileSystem.get(conf);
		
		if (fs.exists(outputPath)) {
			
			fs.delete(outputPath, true);
			
		}
		
		// ①创建Job
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(WCDriver.class);
		
		
		// 为Job创建一个名字
		job.setJobName("wordcount");
		
		// ②设置Job
		// 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
		job.setMapperClass(WCMapper.class);
		job.setReducerClass(WCReducer.class);
		
		// Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
		// 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		/**
		 * 声明使用NLineInputFormat
 		 */
		job.setInputFormatClass(NLineInputFormat.class);
		
		// 设置输入目录和输出目录
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		// ③运行Job
		job.waitForCompletion(true);
		
		
	}

}
